Source code for hysop.topology.topology_descriptor

# Copyright (c) HySoP 2011-2024
#
# This file is part of HySoP software.
# See "https://particle_methods.gricad-pages.univ-grenoble-alpes.fr/hysop-doc/"
# for further info.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


from abc import ABCMeta, abstractmethod
from hysop.tools.htypes import check_instance
from hysop.topology.topology import Topology, TopologyView
from hysop.mesh.mesh import Mesh


[docs] class TopologyDescriptor(metaclass=ABCMeta): """ Describes how a topology should be built. Multiple compatible topology descriptors are gathered during operator graph building and are replaced by a single unique topology upon initialization. """ __slots__ = ("_mpi_params", "_domain", "_backend", "_extra_kwds") def __init__(self, mpi_params, domain, backend, **kwds): """ Initialize a TopologyDescriptor. Notes ----- kwds allows for backend specific variables. CartesianTopologyDescriptor is immutable. """ super().__init__() self._mpi_params = mpi_params self._domain = domain self._backend = backend self._extra_kwds = frozenset(kwds.items()) if ("cl_env" in kwds) and (kwds["cl_env"] is not None): assert kwds["cl_env"].mpi_params is mpi_params def _get_mpi_params(self): """Get mpi parameters.""" return self._mpi_params def _get_domain(self): """Get domain.""" return self._domain def _get_backend(self): """Get backend.""" return self._backend def _get_dim(self): """Get domain dimension.""" return self.domain.dim def _get_extra_kwds(self): """Get extra keyword arguments.""" return dict(self._extra_kwds) mpi_params = property(_get_mpi_params) domain = property(_get_domain) backend = property(_get_backend) extra_kwds = property(_get_extra_kwds) dim = property(_get_dim)
[docs] @staticmethod def build_descriptor(backend, operator, field, handle, **kwds): """ Generate a descriptor from a lower level representation. If handle is already a Topology or a TopologyDescriptor it is returned unchanged. If handle is a CartesianTopologyDescriptors (ie. currently a Discretization), a CartesianTopologyDescriptor is created and returned. Every new topology type should be registered here. """ from hysop.topology.cartesian_descriptor import ( CartesianTopologyDescriptor, CartesianTopologyDescriptors, ) if isinstance(handle, Topology): # handle is already a Topology, so we return it. return handle elif isinstance(handle, TopologyView): return handle._topology elif isinstance(handle, TopologyDescriptor): # handle is already a TopologyDescriptor, so we return it. return handle elif isinstance(handle, CartesianTopologyDescriptors): return CartesianTopologyDescriptor.build_descriptor( backend, operator, field, handle, **kwds ) elif handle is None: # this topology will be determined later return None else: msg = "Unknown handle of class {} to build a TopologyDescriptor." msg = msg.format(handle.__class__) raise TypeError(msg)
[docs] def choose_or_create_topology(self, known_topologies, **kwds): """ Returns a topology that is either taken from known_topologies, a set of user specified topologies which are ensured to be compatible with the current TopologyDescriptor, or created from the descriptor if choose_topology() returns None. """ check_instance(known_topologies, set, values=Topology) topo = self.choose_topology(known_topologies, **kwds) if topo is None: topo = self.create_topology(**kwds) return topo
[docs] @abstractmethod def choose_topology(self, known_topologies, **kwds): """ Find optimal topology parameters from known_topologies. If None is returned, create_topology will be called instead. """ pass
[docs] @abstractmethod def create_topology(self, **kwds): """ Build a topology with the current TopologyDescriptor. """ pass
[docs] def match(self, other, invert=False): """Test if this descriptor is equivalent to the other one.""" if not isinstance(other, TopologyDescriptor): return NotImplemented eq = self.domain is other.domain eq &= self.mpi_params == other.mpi_params eq &= self.backend == other.backend eq &= self.extra_kwds == other.extra_kwds if invert: return not eq else: return eq
@abstractmethod def __eq__(self, other): return self.match(other) @abstractmethod def __ne__(self, other): return not self.match(other, invert=True) @abstractmethod def __hash__(self): h = id(self.domain) h ^= hash(self.mpi_params) h ^= hash(self.backend) h ^= hash(self._extra_kwds) return h @abstractmethod def __str__(self): pass
TopologyDescriptors = (Topology, TopologyDescriptor, type(None)) """ Instance of those types can be used to create a TopologyDescriptor. Thus they can be passed in the variables of each operator. """